package org.shj.spark.streaming

import org.apache.spark._
import org.apache.spark.streaming._

/**
 * input stream 会对应到一个Receiver， Receiver运行在一个Core中，Receiver需要不停地监听去读取数据，所以
 * 分配给SparkStreaming的Core会被长时间占用，直到stop掉这个Receiver。
 * 注意： 如果使用local[1]，则表示只有一个线程，那么这个线程将会被分配给Receiver，此时将没有线程可以用来分配给处理数据！！
 * 
 * 
 * 此例在本地运行时，打印不出信息。 打成jar包后，通过./bin/spark-submit 去提交执行时，可以在控制台看到打印的信息
 * 
 * 在linux 下运行：  nc -lk 9999 可以作为此程序所需数据的输入源
 */
object SocketStreamDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SocketStreamDemo")
    //conf.setMaster("local")
    val sc = new SparkContext(conf)
    
    // Seconds(10) 每隔10秒提交一个任务。 任务数据是这个时间间隔内传过来的数据。
    val ssc = new StreamingContext(sc, Seconds(10)) 
    
    val lines = ssc.socketTextStream("bigdata1", 9999)
    val words = lines.flatMap(_.split("\\s+"))
    val counts = words.map((_, 1)).reduceByKey(_+_)
    
    counts.print()
    
    /**
     * foreachRDD 常用于把每个RDD中的数据推送到外部的System，比如：存成文件或通过network存到数据库
     * 注意 1. foreachRDD中的func参数是一个Function，在运行Streaming应用的driver进程中执行。
     *     2. 通常此func中需要有RDD的action操作，这样才会触发Job
     *     3. 如果向外写数据，代码如下
     *     counts.foreachRDD(rdd => {
            val connect = createConnection()  // 运行在Driver端
            rdd.foreach(record => 
              connect.send(record))  // 在Worker中执行
          })
     *    上面的代码将在Driver端创建连接，而发送的代码是在worker中执行的，这就要求connect可以被序列化，
     *    这样才能把connect传到Worker端。于是可能会想到如下的代码
     *    counts.foreachRDD(rdd => {
            rdd.foreach(record => 
            	val connect = createConnection()  
              connect.send(record))  
          })
     *    上面的代码又会产生另外一个问题： 每条记录都会创建一个Connection！！！正确的做法如下：
     *    counts.foreachRDD(rdd => {
            rdd.foreachPartition(partitionRecord => 
            	val connect = createConnection()  
              partitionRecord.foreach(record => connect.send(record)) )
              connect.close()
          })
                      这样的话，在每个Parition里面，只会创建一次连接。也可以使用连接池获得连接
     */
    counts.foreachRDD(rdd => {
      rdd.sortByKey(false)
    })
    
    ssc.start() //在调用start()之前上面代码不会真正执行
    println("============= after start() =================")
    ssc.awaitTermination()
    
    //awaitTermination()方法会阻塞线程，下面的语句不会被执行，除非上面发生异常或者在提交任务的窗口按 Ctrl+c 停止时，才会执行下面
    // 官网提到手动去stop，应该是在另外线程中调用 ssc.stop()
    println("============= after awaitTermination()=================") 
    ssc.stop(false) //false 表示只停止 StreamingContext， 而不停止 SparkContext
  }
}